package com.zlh.config;

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author zlhuang9
 * @Description
 * @Date 2024-06-18
 */
@Configuration
public class RocketMQPullConsumerConfig {
    @Value("${rocketmq.name-server}")
    private String nameServer;

    @Autowired
    RocketMQProperties rocketMQProperties;

    private final ApplicationContext applicationContext;
    private DefaultLitePullConsumer consumer;

    public RocketMQPullConsumerConfig(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

//    @Bean
//    @Scope("prototype")
//    public DefaultLitePullConsumer defaultLitePullConsumer() {
//        return new DefaultLitePullConsumer();
//    }

    @PostConstruct
    public void start() throws MQClientException, NoSuchMethodException {
        String[] beanNames = applicationContext.getBeanNamesForAnnotation(RocketMQPullConsumer.class);
        for (String beanName : beanNames) {
            Object bean = applicationContext.getBean(beanName);
            Method onMessage = bean.getClass().getDeclaredMethod("onMessage", List.class);
            RocketMQPullConsumer annotation = bean.getClass().getAnnotation(RocketMQPullConsumer.class);
            String topic = annotation.topic();
            String consumerGroup = annotation.consumerGroup();

            DefaultMQPushConsumer finalConsumer;
            if (rocketMQProperties.getConsumer().getAccessKey() != null && rocketMQProperties.getConsumer().getSecretKey() != null) {
                finalConsumer = new DefaultMQPushConsumer(new AclClientRPCHook(
                        new SessionCredentials(rocketMQProperties.getConsumer().getAccessKey(),
                        rocketMQProperties.getConsumer().getSecretKey())));
            } else {
                finalConsumer = new DefaultMQPushConsumer();
            }
            finalConsumer.setNamesrvAddr(rocketMQProperties.getNameServer());
            finalConsumer.setConsumerGroup(consumerGroup);
            System.err.println("topic: " + topic + ", consumerGroup: " + consumerGroup);
            finalConsumer.subscribe(topic, "*");
            finalConsumer.setPullBatchSize(50);
            finalConsumer.setConsumeMessageBatchMaxSize(50);
            // 拉取间隔
            finalConsumer.setPullInterval(10000);

            finalConsumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    // 处理消息
//                    for (MessageExt msg : msgs) {
//                        System.out.printf("Message: %s%n", new String(msg.getBody()));
//                    }
                    try {
                        onMessage.invoke(bean, msgs);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            finalConsumer.start();

//            ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
//
//            // 拉取间隔，单位为毫秒
//            long pullInterval = 5000; // 例如每2秒拉取一次
//
//            // 定时任务，定期拉取消息
//            scheduler.scheduleAtFixedRate(() -> {
//                try {
//                    List<MessageExt> messages = finalConsumer.;
//                    onMessage.invoke(bean, messages);
//                } catch (Exception e) {
//                    e.printStackTrace();
//                }
//            }, 0, pullInterval, TimeUnit.MILLISECONDS);
        }
    }

    @PreDestroy
    public void shutdown() {
        if (consumer != null) {
            consumer.shutdown();
        }
    }
}
